package com.foreveross.proxyip.core.zk;

import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

import org.apache.commons.lang3.StringUtils;
import org.apache.log4j.Logger;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

import com.foreveross.proxyip.core.pools.PoolsManager;


public abstract class ZkLock implements Watcher {
	protected static final Logger log = Logger.getLogger(ZkLock.class);
	 private ZooKeeper zk;
	 private String hosts = null;
	 private String lockName = null;
	 private String nodePath = null;
	 private String nodeName = null;
	 private boolean extra = false;
	 private Integer SESSION_TIMEOUT = 5000;
	 private String waitPath = null;
	 public CountDownLatch latch = new CountDownLatch(1);
	 
	 public ZkLock(String lockName,ZooKeeper zk) throws Exception{
		 this.lockName = StringUtils.contains(lockName, "/")?lockName:"/"+lockName;
		 if(zk==null){
			  zk = new ZooKeeper(hosts, SESSION_TIMEOUT,this);
		 }
		 this.zk = zk;
	 }

	private void init(ZooKeeper zk) throws KeeperException,
			InterruptedException {
		if(zk.exists(this.lockName, false)==null){
				zk.create(this.lockName, this.lockName.getBytes()
			, Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
			}
			nodePath = zk.create(this.lockName + "/lock-", new byte[0],Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);
			nodeName = StringUtils.substringAfter(nodePath, this.lockName+"/");
	}
	 
	 public ZkLock(String lockName) throws Exception{
		 	this(lockName,null);
	 } 
	@Override
	public void process(WatchedEvent event) {
		latch.countDown();
		if (extra&&event.getType() == EventType.NodeDeleted && event.getPath().equals(waitPath)) {
			action();
        }
	}
	
	private Object exec(Long timeout) throws Exception{
		Long begin = System.currentTimeMillis();
		while(true&&System.currentTimeMillis()-begin<timeout){
			List<String> list = zk.getChildren(lockName, false);
			String[] nodes = list.toArray(new String[list.size()]); 
			Arrays.sort(nodes);
			log.info("~~~~~~~~~~~lock nodeName "+nodeName+":::::::"+nodes[0]);
			if(StringUtils.equals(nodeName, nodes[0])){
				log.info("lock exec~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~");
				Object obj = action();
				return obj;
			}else{//等待最小的节点执行完删除
				Stat stat = zk.exists(lockName + "/" + nodes[0],true);
				if(stat!=null){
					latch.await();
				}
//				return waitLock(nodes[0],timeout);
			}
		}
		Object obj = action();
		return obj;
	}
	
//	private Object waitLock(String lower,Long timeout) throws Exception{
//		Stat stat = zk.exists(lockName + "/" + lower,true);
//		if(stat!=null){
//			latch.await(); 
//		}
//			return this.exec(timeout);
//	}
	
	//这里的锁，主线程后面要加Thread.sleep()
	private void extraExec() throws Exception{
        latch.await();
        List<String> childrenNodes = zk.getChildren(lockName, false);
        if (childrenNodes.size() == 1) {
        	action();
        } else {
        	  Collections.sort(childrenNodes);
              int index = childrenNodes.indexOf(nodeName);
            if (index == -1) {
                // never happened
            } else if (index == 0) {
            	action();
            } else {
//            	log.debug("waitPath:"+childrenNodes.get(index - 1));
            	this.waitPath = this.lockName + "/" + childrenNodes.get(index - 1);
                // 在waitPath上注册监听器, 当waitPath被删除时, zookeeper会回调监听器的process方法
            	 if(zk.exists(waitPath, true)==null){
            		action();
                 }
            }
        }
	}
	
	private Object action() {
		try{
		return doAction();
		} catch(Exception e){e.printStackTrace();
			throw new RuntimeException(e);
		}finally{
			try {
				
				zk.delete(nodePath, -1);
				nodePath = null;
				nodeName = null;
//				zk.close();
			} catch (Exception e) {
				throw new RuntimeException(e);
			}
			
		}
	}
	
	/**
	 * 主程必须守候,因为是与脱离主线程，所以无返回值。
	 * @return
	 * @throws Exception
	 */
	public void extraExecute() throws Exception{
		
			init(zk);
			this.extra = true;
			this.extraExec();
	}
	
	public Object execute(Long timeout) throws Exception{
			Long start = System.currentTimeMillis();
			while((nodePath!=null||nodeName!=null)&&System.currentTimeMillis()-start<timeout){
				if(System.currentTimeMillis()-start>=timeout){
					//该应用有程序在执行该锁，等待正在执行的应用执行完成，如果超时；退出不执行,并且设置
					nodePath = null;
					nodeName = null;
					log.info("other task exec timeout");
					return null;
				}
			}
			init(zk);
			log.debug("lock~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~");
			return exec(timeout);
	}
	
	public abstract Object doAction() throws Exception ;
	
	
	public static void main(String[] args) throws Exception {
		Object obj = null;
//		final CountDownLatch startSignal = new CountDownLatch(10); 
        for (int i = 0; i < 10; i++) {
        	final int a = i;
        	ZkLock lock = new ZkLock("test") {
				
				@Override
				public Object doAction() throws Exception {
					Thread.sleep(Long.MAX_VALUE);
					return null;
//					log.debug(a);
//					return null;
				}
			};
			lock.execute(100000L);
//            new Thread() {
//                public void run() {
//                	ZkLock lock;
//					try {
//						lock = new ZkLock("test") {
//							@Override
//							public Object doAction() {
//								return null;
//							}
//						};
////						startSignal.await();
//						lock.execute();
//						//lock.extraExecute();
//					} catch (Exception e) {
//						// TODO Auto-generated catch block
//						e.printStackTrace();
//					}
//                }
//            }.start();
//            startSignal.countDown();
        }
	}

}
